package com.qqt.csr.im.service;

import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.corundumstudio.socketio.SocketIOClient;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.qqt.csr.common.utils.CacheKeyUtil;
import com.qqt.csr.common.utils.JsonUtil;
import com.qqt.csr.common.utils.RedissonLockUtil;
import com.qqt.csr.common.utils.TransactionUtil;
import com.qqt.csr.im.assembler.ConnectionAssembler;
import com.qqt.csr.im.entity.Connection;
import com.qqt.csr.im.enums.ConnectStateEnum;
import com.qqt.csr.im.mapper.ConnectionMapper;
import com.qqt.csr.im.socketio.utils.SocketIOLocalCacheUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.redisson.api.BatchResult;
import org.redisson.api.RBatch;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @Author: yang hao
 * @Date: 2024/10/14 9:50
 * @Description: 人工客服 im 长连接信息 业务层
 **/
@Slf4j
@Service
@SuppressWarnings("all")
public class ConnectionService extends ServiceImpl<ConnectionMapper, Connection> {
    /**
     * 在线本地缓存连接，key:accountId
     */
    private Cache<Long, List<Connection>> ONLINE_CONNECT_CACHE = Caffeine.newBuilder()
            .maximumSize(10000)
            .expireAfterAccess(10, TimeUnit.MINUTES)
            .build();
    @Autowired
    private AccountService accountService;
    @Autowired
    private RedissonLockUtil redissonLockUtil;
    @Autowired
    private TransactionUtil transactionUtil;
    @Autowired
    private RedissonClient redissonClient;

    /**
     * 创建或更新
     *
     * @param client
     * @param accountId
     */
    public Connection connect(SocketIOClient client) {
        Long accountId = client.get("accountId");
        String accountUserId = client.get("accountUserId");
        String deviceFlag = client.get("deviceFlag");
        Integer deviceType = Integer.parseInt(client.get("deviceType"));
        Connection cnn = transactionUtil.execute(() -> {
            Connection connection = ConnectionAssembler.build(deviceFlag, deviceType, accountId);
            // 判断连接是否存在，不存在则创建，存在则更改
            Connection dbConn = this.lambdaQuery().select(Connection::getId)
                    .eq(Connection::getAccountId, accountId)
                    .eq(Connection::getDeviceFlag, deviceFlag)
                    .eq(Connection::getDeviceType, deviceType).one();
            if (Objects.nonNull(dbConn)) {
                connection.setId(dbConn.getId());
                connection.setCreateTime(null);
            }
            saveOrUpdate(connection);
            accountService.updateOnline(accountId, accountUserId);
            return connection;
        });
        clearCache(accountId);
        return cnn;
    }

    /**
     * 断开连接
     *
     * @param connection
     */
    public void disconnect(SocketIOClient client) {
        Connection connection = client.get("connection");
        if (connection == null) {
            return;
        }
        transactionUtil.execute(() -> {
            connection.setState(ConnectStateEnum.DISCONNECT.getCode());
            connection.setDisconnectionTime(new Date());
            this.lambdaUpdate().set(Connection::getState, connection.getState())
                    .set(Connection::getDisconnectionTime, connection.getDisconnectionTime())
                    .set(Connection::getUpdateTime, new Date())
                    .eq(Connection::getId, connection.getId()).update();

            List<Connection> cnnList = this.getOnlineByAccountId(connection.getAccountId());
            cnnList.removeIf(cnn -> cnn.getId().equals(connection.getId()));
            if (CollectionUtils.isEmpty(cnnList)) {
                String accountUserId = client.get("accountUserId");
                accountService.updateOffline(connection.getAccountId(), accountUserId);
            }
        });
        clearCache(connection.getAccountId());
    }

    /**
     * 断开连接
     *
     * @param connection
     */
    public void disconnect(List<Connection> connectionList) {
        if (CollectionUtils.isEmpty(connectionList)) {
            return;
        }
        List<Long> connetionIdList = connectionList.stream().map(Connection::getId).collect(Collectors.toList());
        List<Long> accountIdList = connectionList.stream().map(Connection::getAccountId).distinct().collect(Collectors.toList());
        transactionUtil.execute(() -> {
            this.lambdaUpdate().set(Connection::getState, ConnectStateEnum.DISCONNECT.getCode())
                    .set(Connection::getDisconnectionTime, new Date())
                    .set(Connection::getUpdateTime, new Date())
                    .in(Connection::getId, connetionIdList).update();

            Map<Long, List<Connection>> onlineAccountMap = this.getOnlineByAccountId(accountIdList);
            List<Long> offlineAccountIdList = new ArrayList<>();
            for (Long accountId : accountIdList) {
                List<Connection> cnnList = onlineAccountMap.get(accountId);
                cnnList.removeIf(cnn -> connetionIdList.contains(cnn.getId()));
                if (CollectionUtils.isEmpty(cnnList)) {
                    offlineAccountIdList.add(accountId);
                }
            }
            accountService.updateOffline(offlineAccountIdList);
        });

        clearCache(accountIdList);
    }

    /**
     * 根据账户id获取所有连接
     *
     * @param accountId
     * @return
     */
    public List<Connection> getOnlineByAccountId(Long accountId) {
        if (accountId == null) {
            return Lists.newArrayList();
        }
        List<Connection> cnnList = ONLINE_CONNECT_CACHE.getIfPresent(accountId);
        if (cnnList != null) {
            return cnnList;
        }

        String redisKey = String.format(CacheKeyUtil.IM.ACCOUNT_CNN_LIST_KEY, accountId);
        cnnList = Optional.ofNullable(redissonClient.getBucket(redisKey).get()).map(Object::toString)
                .map(o -> JsonUtil.toList(o, Connection.class)).orElse(null);
        if (cnnList != null) {
            ONLINE_CONNECT_CACHE.put(accountId, cnnList);
            return cnnList;
        }

        cnnList = this.lambdaQuery().select(Connection::getId, Connection::getDeviceFlag, Connection::getDeviceType,
                        Connection::getAccountId, Connection::getState, Connection::getServiceId)
                .eq(Connection::getAccountId, accountId).list()
                .stream()
                .filter(item -> ConnectStateEnum.CONNECT.getCode().equals(item.getState()))
                .collect(Collectors.toList());
        ONLINE_CONNECT_CACHE.put(accountId, cnnList);
        redissonClient.getBucket(redisKey).setAsync(JsonUtil.toJson(cnnList), CacheKeyUtil.EXPIRE_TIME_8H, TimeUnit.MILLISECONDS);
        return cnnList;
    }

    /**
     * 根据账户id获取所有在线连接
     *
     * @param accountId
     * @return key:accountId
     */
    public Map<Long, List<Connection>> getOnlineByAccountId(List<Long> accountIdList) {
        if (CollectionUtils.isEmpty(accountIdList)) {
            return Maps.newHashMap();
        }

        Map<Long, List<Connection>> connectionMap = new HashMap<>();
        if (accountIdList.size() == 1) {
            connectionMap.put(accountIdList.get(0), getOnlineByAccountId(accountIdList.get(0)));
            return connectionMap;
        }

        // 本地缓存获取
        List<Long> needQueryRedisAccountIdList = new ArrayList<>();
        accountIdList.forEach(accountId -> {
            List<Connection> localCacheCnnList = ONLINE_CONNECT_CACHE.getIfPresent(accountId);
            if (localCacheCnnList != null) {
                connectionMap.put(accountId, localCacheCnnList);
            } else {
                needQueryRedisAccountIdList.add(accountId);
            }
        });
        if (CollectionUtils.isEmpty(needQueryRedisAccountIdList)) {
            return connectionMap;
        }

        // redis缓存获取
        Lists.partition(needQueryRedisAccountIdList, 100).forEach(pList -> {
            RBatch redisBatch = redissonClient.createBatch();
            pList.forEach(accountId -> {
                String redisKey = String.format(CacheKeyUtil.IM.ACCOUNT_CNN_LIST_KEY, accountId);
                redisBatch.getBucket(redisKey).getAsync();
            });
            BatchResult batchResult = redisBatch.execute();
            List<Object> objectList = batchResult.getResponses();
            for (int i = 0; i < pList.size(); i++) {
                Long accountId = pList.get(i);
                Object cnnObj = objectList.get(i);
                if (cnnObj != null) {
                    List<Connection> cnnList = JsonUtil.toList(cnnObj.toString(), Connection.class);
                    if (CollectionUtils.isNotEmpty(cnnList)) {
                        accountId = cnnList.get(0).getAccountId();
                    }
                    connectionMap.put(accountId, cnnList);
                    ONLINE_CONNECT_CACHE.put(accountId, cnnList);
                }
            }
        });

        // 数据库查询
        List<Long> needQueryDbAccountIdList = needQueryRedisAccountIdList.stream()
                .filter(accountId -> !connectionMap.containsKey(accountId)).collect(Collectors.toList());
        if (CollectionUtils.isEmpty(needQueryDbAccountIdList)) {
            return connectionMap;
        }

        Map<Long, List<Connection>> dbMap = this.lambdaQuery().select(
                        Connection::getId, Connection::getDeviceFlag, Connection::getDeviceType,
                        Connection::getAccountId, Connection::getState, Connection::getServiceId)
                .in(Connection::getAccountId, needQueryDbAccountIdList)
                .list()
                .stream()
                .filter(item -> ConnectStateEnum.CONNECT.getCode().equals(item.getState()))
                .collect(Collectors.groupingBy(Connection::getAccountId));

        Lists.partition(needQueryDbAccountIdList, 100).forEach(pList -> {
            RBatch redisBatch = redissonClient.createBatch();
            pList.forEach(accountId -> {
                List<Connection> cnnList = Optional.ofNullable(dbMap.get(accountId)).orElseGet(Lists::newArrayList);
                connectionMap.put(accountId, cnnList);
                ONLINE_CONNECT_CACHE.put(accountId, cnnList);
                String redisKey = String.format(CacheKeyUtil.IM.ACCOUNT_CNN_LIST_KEY, accountId);
                redisBatch.getBucket(redisKey).setAsync(JsonUtil.toJson(cnnList), CacheKeyUtil.EXPIRE_TIME_8H, TimeUnit.MILLISECONDS);
            });
            redisBatch.execute();
        });
        return connectionMap;
    }

    private void clearCache(List<Long> accountIdList) {
        if (CollectionUtils.isEmpty(accountIdList)) {
            return;
        }
        RBatch redisBatch = redissonClient.createBatch();
        accountIdList.forEach(accountId -> {
            ONLINE_CONNECT_CACHE.invalidate(accountId);
            String redisKey = String.format(CacheKeyUtil.IM.ACCOUNT_CNN_LIST_KEY, accountId);
            redisBatch.getBucket(redisKey).deleteAsync();
        });
        redisBatch.execute();
    }

    private void clearCache(Long accountId) {
        if (accountId == null || accountId <= 0) {
            return;
        }
        ONLINE_CONNECT_CACHE.invalidate(accountId);
        String redisKey = String.format(CacheKeyUtil.IM.ACCOUNT_CNN_LIST_KEY, accountId);
        redissonClient.getBucket(redisKey).deleteAsync();
    }

}
